package com.jacky.iot.manager;

import io.netty.handler.codec.mqtt.MqttQoS;
import io.vertx.core.buffer.Buffer;
import io.vertx.mqtt.MqttEndpoint;
import io.vertx.mqtt.MqttTopicSubscription;

import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;

public class EndpointManager {

    private final Map<String, MqttEndpoint> clients = new ConcurrentHashMap<>();
    private final Map<String, Set<String>> clientSubscriptions = new ConcurrentHashMap<>();
    private final Map<String, Set<String>> topicSubscribers = new ConcurrentHashMap<>();

    public void addClient(String clientId, MqttEndpoint endpoint) {
        clients.put(clientId, endpoint);
    }

    public void removeClient(String clientId) {
        // 移除客户端时清理所有订阅关系
        Set<String> subscriptions = clientSubscriptions.get(clientId);
        if (subscriptions != null) {
            subscriptions.forEach(topic -> {
                topicSubscribers.getOrDefault(topic, Set.of())
                        .remove(clientId);
            });
        }
        clients.remove(clientId);
        clientSubscriptions.remove(clientId);
    }

    public void addSubscription(String clientId, MqttTopicSubscription topic) {
        clientSubscriptions.computeIfAbsent(clientId, k -> new CopyOnWriteArraySet<>()).add(topic.topicName());
        topicSubscribers.computeIfAbsent(topic.topicName(), k -> new CopyOnWriteArraySet<>()).add(clientId);
    }

    // TODO 这里要考虑一下如果一个主题已经没有订阅者了,是不是应该清除掉,不要留着
    public void removeSubscription(String clientId, String topic) {
        clientSubscriptions.getOrDefault(clientId, Set.of()).remove(topic);
        topicSubscribers.getOrDefault(topic, Set.of()).remove(clientId);
    }

    public MqttEndpoint getClient(String clientId) {
        return clients.get(clientId);
    }

    public Map<String, MqttEndpoint> getClients() {
        return clients;
    }

    public void publishToTopic(String topic, Buffer payload, MqttQoS qos, boolean isDup, boolean retain) {
        // 仅发送给订阅了匹配主题的客户端
        // todo 这里现在是精准匹配订阅主题,要做模糊匹配的主题也要转发消息
        topicSubscribers.getOrDefault(topic, Set.of()).forEach(clientId -> {
            MqttEndpoint endpoint = clients.get(clientId);
            if (endpoint != null) {
                endpoint.publish(topic, payload, qos, isDup, retain);
            }
        });
    }
}
